package window;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

/**
 * kafka读取数据 生成watermark
 * windowAll 每一个分区满足触发条件 窗口才会触发
 */
public class KafkaSourceWaterMarkDemo {

    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        // 配置
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers","master:9092,slave1:9092,slave2:9092");
        properties.setProperty("group.id","test");
        properties.setProperty("auto.offset.reset","earliest");

        DataStreamSource<String> stream = env.addSource(new FlinkKafkaConsumer<String>("wc", new SimpleStringSchema(), properties));
        // 设置并行度为4
        stream.setParallelism(4);

        // 提取watermark
        SingleOutputStreamOperator<String> watermarks = stream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(0)) {
            @Override
            public long extractTimestamp(String s) {
                return Long.parseLong(s.split(",")[0]);
            }
        });

        // map
        SingleOutputStreamOperator<Tuple2<Long, Integer>> mapped = watermarks.map(new MapFunction<String, Tuple2<Long, Integer>>() {
            @Override
            public Tuple2<Long, Integer> map(String s) throws Exception {
                String[] fields = s.split(",");
                return Tuple2.of(Long.parseLong(fields[0]), Integer.parseInt(fields[1]));
            }
        });

        // 划分窗口
        AllWindowedStream<Tuple2<Long, Integer>, TimeWindow> window = mapped.windowAll(TumblingEventTimeWindows.of(Time.seconds(5)));

        // 聚合 window apply
        SingleOutputStreamOperator<Tuple2<Long, Integer>> summed = window.sum(1);

        // sink print
        summed.print();

        env.execute("kafka watermark");
    }
}
